package com.bbx.flink.case1;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.walkthrough.common.sink.AlertSink;
import org.apache.flink.walkthrough.common.entity.Alert;
import org.apache.flink.walkthrough.common.entity.Transaction;
import org.apache.flink.walkthrough.common.source.TransactionSource;

/**
 * 反欺诈demo
 * 欺诈检测器应针对任何进行小额交易然后立即进行大笔交易的帐户输出警报。小的小于$ 1.00，大的大于$ 500。
 * 假设您的欺诈检测器为特定帐户处理以下交易流。
 * 交易3和4应该标记为欺诈，因为这是一笔小交易，0.09美元，然后是一笔大交易，510美元。另外，
 * 交易7、8和9也不是欺诈，因为少量的$ 0.02不会立即跟随大笔的交易；而是有一个中间事务破坏了模式。
 *
 * 为此，欺诈检测者必须记住事件之间的信息。大型交易只有在前一个交易很小的情况下才是欺诈性的。
 * 记住事件之间的信息需要状态，这就是为什么我们决定使用KeyedProcessFunction的原因。
 * 它提供了对状态和时间的细粒度控制，这将使我们能够在整个演练中发展出具有更复杂要求的算法。
 *
 *
 *
 *
 */
public class FraudDetectionJob {


    public static void main(String[] args) throws Exception {
        //设置执行环境，执行环境用来设置job属性，创建源，触发job 执行
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //创建源， 源将外部系统（kafka，rocketMq等）的数据摄取到Flink Jobs 中
        DataStream<Transaction> transactions = env
                .addSource(new TransactionSource())
                .name("transactions");
        //划分事件和检测欺诈
        //keyBy    划分事件
        //process   对事件进行处理
        DataStream<Alert> alerts = transactions
                .keyBy(Transaction::getAccountId)
                .process(new FraudDetector())
                .name("fraud-detector");
        //输出结果，将结果输出到外部系统（kafka等）
        alerts
                .addSink(new AlertSink())
                .name("send-alerts");
        // 执行job
        env.execute("Fraud Detection");


    }}
